package com.xyy.saas.payment.util.mq;

import com.xyy.saas.payment.common.RocketMqGroup;
import com.xyy.saas.payment.common.constants.MqConstant;
import com.xyy.saas.payment.cores.bo.QueryResultBo;
import com.xyy.saas.payment.cores.vo.ResultVO;
import com.xyy.saas.payment.query.core.service.QueryTradeService;
import com.xyy.saas.payment.common.Constants;
import com.xyy.saas.payment.common.RocketMqTopic;
import com.xyy.saas.payment.dao.mapper.WithdrawOrderMapper;
import com.xyy.saas.payment.dao.model.WithdrawOrder;
import com.xyy.saas.payment.util.CallbackUtil;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.common.UtilAll;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

/**
 * @author wangtianqi
 * @date 2019/11/19 20:09
 */
@Service
@RocketMQMessageListener(topic = RocketMqTopic.TOPIC_WITHDRAW_CHECK, consumerGroup = RocketMqGroup.GROUP_WITHDRAW_CHECK, consumeMode = ConsumeMode.ORDERLY)
public class WithdrawCheckConsumer implements RocketMQListener<MessageExt>, RocketMQPushConsumerLifecycleListener {

    private static final Logger LOGGER = LoggerFactory.getLogger(RetryNotifyConsumer.class);

    @Autowired
    MqProducer mqProducer;

    @Autowired
    CallbackUtil callbackUtil;

    @Autowired
    QueryTradeService queryTradeService;

    @Autowired
    WithdrawOrderMapper withdrawOrderMapper;


    @Override
    public void onMessage(MessageExt message) {
        String body = new String(message.getBody());
        try {
            String tags = message.getTags();
            int delayLevel = message.getDelayTimeLevel();
            LOGGER.info("WithdrawCheckConsumer#onMessage tags:{} body:{}", tags, body);
            WithdrawOrder withdrawOrder = withdrawOrderMapper.selectByWithdrawNo(body);
            if (withdrawOrder.getStatus().equals(WithdrawOrder.STATUS_IN)) {
                int count = getCurrentCount(delayLevel);
                if (count <= MqConstant.DELAY_LEVELS.length - 1) {
                    count++;
                    // 检查提现
                    ResultVO<QueryResultBo> resultVO = queryTradeService.queryWithdraw(body);
                    if(QueryResultBo.STATUS_PROCESSING.equals(resultVO.getResult().getStatus())) {
                        mqProducer.syncSendDelayMsg(RocketMqTopic.TOPIC_WITHDRAW_CHECK, body, count);
                    } else {
                        LOGGER.info("WithdrawCheckConsumer#onMessage check withdraw finish, withdrawNo:{}", body);
                    }
                } else {
                    LOGGER.error("WithdrawCheckConsumer#onMessage check withdraw count max, withdrawNo:{}", body);
                }
            }
        } catch (Exception e) {
            LOGGER.error("WithdrawCheckConsumer#onMessage consumer error body:{}",body, e);
        }
    }

    private static int getCurrentCount(int currentLevel) {
        for (int level = 0; level <= MqConstant.WITHDRAW_DELAY_LEVELS.length - 1; level++) {
            if (MqConstant.WITHDRAW_DELAY_LEVELS[level] == currentLevel) {
                return level;
            }
        }
        return 0;
    }


    @Override
    public void prepareStart(DefaultMQPushConsumer consumer) {
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        consumer.setConsumeTimestamp(UtilAll.timeMillisToHumanString3(System.currentTimeMillis()));
    }
}